package org.dragonnova.iot.business.cloud.mq.service;

import java.util.Map;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.dragonnova.iot.business.cloud.mq.service.KafkaService.IProducerFactory;

/**
 * 
 * @description: 消息生产工厂
 * @author songxy DateTime 2017年5月27日 上午11:50:06
 * @company winter
 * @email thinkdata@163.com
 * @version 1.0
 */
public class KafkaProducerFactory {

	private KafkaProducer producer;

	private IProducerFactory producerFactory;

	private Map<String, Object> configs;

	public KafkaProducer<String, String> getProducer() {
		if (producer == null) {
			synchronized (this) {
				producer = producerFactory.createKafkaProducer();
			}
		}
		return producer;
	}

	/**
	 * @return get configs
	 */

	public Map<String, Object> getConfigs() {
		return configs;
	}

	/**
	 * @param configs
	 *            set configs
	 */

	public void setConfigs(Map<String, Object> configs) {
		this.configs = configs;
	}

	public void start() {
		producerFactory = new ProducerFactory(configs);
	}

	public void stop() {
		if (producer != null) {
			producer.close();
		}
	}

	private class ProducerFactory<K, V> implements IProducerFactory {
		private Map<String, Object> configs;

		private ProducerFactory(Map<String, Object> configs) {
			this.configs = configs;
		}

		@Override
		public KafkaProducer<K, V> createKafkaProducer() {
			return new KafkaProducer<K, V>(this.configs);
		}

	}

}
